/**
 * 
 */
package stc.skymobi.fsm.core;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import javax.management.ObjectName;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.Closure;
import org.apache.commons.collections.Predicate;
import org.apache.commons.collections.Transformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.MBeanExporter;

import stc.skymobi.ebus.EventBus;
import stc.skymobi.fsm.ContextKeyFactory;
import stc.skymobi.fsm.FSMContext;
import stc.skymobi.fsm.FSMDashboard;
import stc.skymobi.fsm.FiniteStateMachine;
import stc.skymobi.fsm.MutableFSMContext;
import stc.skymobi.fsm.State;
import stc.skymobi.fsm.StateRepository;
import stc.skymobi.fsm.annotation.IndicateInterface;
import stc.skymobi.fsm.annotation.Indicator;
import stc.skymobi.fsm.event.TimeoutEvent;


/**
 * @author hp
 *
 */
public abstract class SmartFSM implements FiniteStateMachine {
	
    private static final Logger logger = 
    	LoggerFactory.getLogger(SmartFSM.class);
    
	private	final Map<Object, SmartContext>	contextHolder
		= new ConcurrentHashMap<Object, SmartContext>();
	
    private	final Map<TimeoutEvent, Long>	timeoutHolder = 
    	new HashMap<TimeoutEvent, Long>();
    
	private final String		name;
	private	final int			fsmId;
    private	volatile boolean	running = true;
	private	volatile ContextKeyFactory keyFactory;
	private volatile Predicate	initeventAcceptor;
	private	volatile String		initStateName;
	private	volatile StateRepository stateRepository;
	private	final ThreadPoolExecutor workService;
    private final ScheduledThreadPoolExecutor innerService;
	
    private volatile EventBus	eventBus;
    private volatile String		eventStateChanged;
    private	volatile boolean	logStateChanged = true;
    private	volatile String		eventOfDestroyContext = null;
    
	private	volatile int		executorPoolSizeLimit = 100;
    private	final AtomicInteger	totalJobCount = new AtomicInteger(0);
    private	final AtomicInteger	activeJobCount = new AtomicInteger(0);
	private	final AtomicInteger	largestActiveJobCount = new AtomicInteger(0);
	private	final AtomicInteger	periodLargestActiveJobCount = new AtomicInteger(0);
	
    private	volatile boolean	checkBusy = true;
    private	volatile int		busyThreshold = 200;
    private	volatile Closure	busyHandler = null;
    private	volatile Transformer snapshotFactory = null;
    private	final AtomicInteger dealHandledCount = new AtomicInteger(0);
    private	final AtomicInteger dealCompletedCount = new AtomicInteger(0);
    private	final AtomicInteger dealBypassCount = new AtomicInteger(0);
    
    //	50L ms
    private	long				intervalOfCheckTimeoutEvent = 50L;
    private	ScheduledFuture<?>	futureOfCheckTimeoutEvent = null;
    
    //	1L s
    private	long				intervalOfAdjustExecutorCorePoolSize = 1 * 1000L;
    private	ScheduledFuture<?>	futureOfAdjustExecutorCorePoolSize = null;
    
    //	dashboard statistics value
    private	volatile long		dealLargestTimeToLive = -1;
    private	volatile long		dealLargestTimeToActive = -1;
    private	volatile long		dealSmallestTimeToLive = Integer.MAX_VALUE;
    private	volatile long		dealSmallestTimeToActive = Integer.MAX_VALUE;
    private	final AtomicLong	dealTotalTimeToLive = new AtomicLong(0);
    private	final AtomicLong	dealTotalTimeToActive = new AtomicLong(0);

    //	deal statistics value
    private	volatile Transformer	transformerOfTTL = null;
    private	volatile Transformer	transformerOfTTA = null;

    private	final Map<Object, AtomicLong>	erCounters = 
    	new HashMap<Object, AtomicLong>();
    
    private static class ERTuple {
    	Transformer				trans;
    	Map<Object, AtomicLong>	counter = new HashMap<Object, AtomicLong>();
    }
    
    private	final Map<String, ERTuple>	erTuplesOfTTL = 
    	new HashMap<String, ERTuple>();
    
    private	final Map<String, ERTuple>	erTuplesOfTTA = 
    	new HashMap<String, ERTuple>();
    
    private final Map<String, ERTuple> erTuplesOfStateAwareTTL = new HashMap<String, ERTuple>();

//    private volatile FSMStatisticer	statisticer;
    
    //	JMX support
    private	volatile MBeanExporter 	mbeanExporter = null;
    
    //	FSM JMX's prefix
    private	volatile String			objectNamePrefix;
    
    private	final List<ObjectName>	objectNameList = new ArrayList<ObjectName>();
    
    private String createObjectName(final String suffix) {
    	return	objectNamePrefix + "," + suffix;
    }
    
    private String createObjectNameForDeal(final Object reason, final String suffix) {
    	return	objectNamePrefix + ",module=deal,result=" + reason.toString() + "," + suffix;
    }
    
    private	final SmartContext.StatusListener ctxStatusListener = 
    	new SmartContext.StatusListener() {
			public void onActive(SmartContext ctx) {
				incActiveJobCount();
			}
	
			public void onUnactive(SmartContext ctx) {
				decActiveJobCount();
			}
	    };
    
	private	void registerMBean(final String on, final Object mbean) {
		final MBeanExporter	exporter = mbeanExporter;
		
		if ( null != exporter ) {
			ObjectName objectName;
			try {
				objectName = new ObjectName(on);
				exporter.registerManagedResource(mbean, objectName);
				objectNameList.add(objectName);
			} catch (Exception e) {
				logger.error("registerMBean", e);
			}
		}
		
	}
	
	private	void unregisterAllMBean() {
		final MBeanExporter	exporter = mbeanExporter;
		
		if ( null != exporter ) {
			for ( ObjectName on : objectNameList ) {
				try {
					exporter.getServer().unregisterMBean(on);
				} catch (Exception e) {
					logger.error("unregisterAllObjectName:", e);
				}
			}
			objectNameList.clear();
		}
	}
	
	public SmartFSM(final String fsmName, final int fsmId) {
    	this.name = fsmName;
    	this.fsmId = fsmId;
    	this.objectNamePrefix = "stc.skymobi:type=smartfsm,name=" 
    			+ this.name + ",id=" + this.fsmId;
		this.workService = 
	        new ThreadPoolExecutor(0, executorPoolSizeLimit, 
                1L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(),
                new ThreadFactory(){
					public Thread newThread(Runnable r) {
						return new Thread(r, name);
					}});
		
		this.innerService = 
			new ScheduledThreadPoolExecutor(1, 
				new ThreadFactory(){
					public Thread newThread(Runnable r) {
						return new Thread(r, name + "-timer");
					}});
    	
		rescheduleCheckTimeoutEvent();
		
		rescheduleAdjustThreadPoolCoreSize();
    }
    
	public void start() {
		registerMBean(createObjectName("module=dashboard"), new DashboardImpl());
		registerMBean(createObjectName("module=fsm"), this);
	}
	
	public void stop() {
		if ( !innerService.isShutdown() ) {
			innerService.shutdownNow();
		}
		
		if ( !workService.isShutdown() ) {
			workService.shutdownNow();
		}
		
		unregisterAllMBean();
		
		running = false;
		
		if ( logger.isInfoEnabled() ) {
			logger.info("fsm {}/{} stopped.", this.name, this.fsmId);
		}
	}
	
	public boolean isRunning() {
		return	running;
	}
	
	/**
	 * @return the eventOfDestroyContext
	 */
	public String getEventOfDestroyContext() {
		return eventOfDestroyContext;
	}

	/**
	 * @param eventOfDestroyContext the eventOfDestroyContext to set
	 */
	public void setEventOfDestroyContext(final String eventOfDestroyContext) {
		this.eventOfDestroyContext = eventOfDestroyContext;
	}

	/**
	 * @return the initState
	 */
	public String getInitStateName() {
		return initStateName;
	}

	/**
	 * @param initState the initState to set
	 */
	public void setInitStateName(final String initState) {
		this.initStateName = initState;
	}
	
	public Transformer getSnapshotFactory() {
		return snapshotFactory;
	}

	public void setSnapshotFactory(final Transformer snapshotFactory) {
		this.snapshotFactory = snapshotFactory;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(final EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public String getEventStateChanged() {
		return eventStateChanged;
	}

	public void setEventStateChanged(final String eventStateChanged) {
		this.eventStateChanged = eventStateChanged;
	}

	public String getName() {
		return name;
	}
	
	public int getExecutorPoolSizeLimit() {
		return executorPoolSizeLimit;
	}

	public void setExecutorPoolSizeLimit(final int poolSize) {
		if ( poolSize > 0 ) {
			this.executorPoolSizeLimit = poolSize;
			this.reducePoolSize(poolSize);
			this.workService.setMaximumPoolSize(poolSize);
		}
	}

	public String getStateRepositoryAsString() {
		return stateRepository.toString();
	}
	
	public StateRepository getStateRepository() {
		return stateRepository;
	}
	
	public void setStateRepository(final StateRepository stateRepository) {
		this.stateRepository = stateRepository;
	}
	
	public ContextKeyFactory getKeyFactory() {
		return keyFactory;
	}
	
	public void setKeyFactory(final ContextKeyFactory keyFactory) {
		this.keyFactory = keyFactory;
	}
	
	/**
	 * @return the initeventAcceptor
	 */
	public Predicate getIniteventAcceptor() {
		return initeventAcceptor;
	}

	/**
	 * @param initeventAcceptor the initeventAcceptor to set
	 */
	public void setIniteventAcceptor(final Predicate initeventAcceptor) {
		this.initeventAcceptor = initeventAcceptor;
	}

	public Closure getBusyHandler() {
		return busyHandler;
	}

	public void setBusyHandler(final Closure busyHandler) {
		this.busyHandler = busyHandler;
	}

	public boolean isLogStateChanged() {
		return logStateChanged;
	}

	public void setLogStateChanged(final boolean logStateChanged) {
		this.logStateChanged = logStateChanged;
	}

	public boolean isCheckBusy() {
		return checkBusy;
	}

	public void setCheckBusy(final boolean checkBusy) {
		this.checkBusy = checkBusy;
	}

	public int getBusyThreshold() {
		return busyThreshold;
	}

	public void setBusyThreshold(final int busyThreshold) {
		this.busyThreshold = busyThreshold;
	}

	/**
	 * @return the transformerOfTimeToLive
	 */
	public Transformer getTransformerOfTimeToLive() {
		return transformerOfTTL;
	}

	/**
	 * @param transformerOfTimeToLive the transformerOfTimeToLive to set
	 */
	public void setTransformerOfTimeToLive(Transformer transformerOfTimeToLive) {
		this.transformerOfTTL = transformerOfTimeToLive;
	}

//	public void setupTransformerOfEndReasonTimeToLive(String endReason, 
//			Transformer transformerOfTimeToLive) {
//		ERTuple tuple = erTuplesOfTTL.get(endReason);
//		
//		if ( null != tuple ) {
//			throw new RuntimeException("EndReason[" + endReason +"]'s transformer already setup!");
//		}
//		
//		tuple = new ERTuple();
//		tuple.trans = transformerOfTimeToLive;
//		
//		erTuplesOfTTL.put(endReason, tuple);
//	}
//	
//	public void setupTransformerOfEndReasonTimeToActive(String endReason, 
//			Transformer transformerOfTimeToActive) {
//		ERTuple tuple = erTuplesOfTTA.get(endReason);
//		
//		if ( null != tuple ) {
//			throw new RuntimeException("EndReason[" + endReason +"]'s transformer already setup!");
//		}
//		
//		tuple = new ERTuple();
//		tuple.trans = transformerOfTimeToActive;
//		
//		erTuplesOfTTA.put(endReason, tuple);
//	}
//	
	/**
	 * @return the transformerOfTimeToActive
	 */
	public Transformer getTransformerOfTimeToActive() {
		return transformerOfTTA;
	}

	/**
	 * @param transformerOfTimeToActive the transformerOfTimeToActive to set
	 */
	public void setTransformerOfTimeToActive(Transformer transformerOfTimeToActive) {
		this.transformerOfTTA = transformerOfTimeToActive;
	}

	/**
	 * @return the fsmId
	 */
	public int getFsmId() {
		return fsmId;
	}

	/**
	 * @param unit
	 * @return
	 * @see java.util.concurrent.ThreadPoolExecutor#getKeepAliveTime(java.util.concurrent.TimeUnit)
	 */
	public long getExecutorKeepAliveTime() {
		if ( null != workService ) {
			return workService.getKeepAliveTime(TimeUnit.MILLISECONDS);
		}
		else {
			return	-1;
		}
	}

	/**
	 * @param time
	 * @param unit
	 * @see java.util.concurrent.ThreadPoolExecutor#setKeepAliveTime(long, java.util.concurrent.TimeUnit)
	 */
	public void setExecutorKeepAliveTime(long time) {
		if ( null != workService ) {
			workService.setKeepAliveTime(time, TimeUnit.MILLISECONDS);
		}
	}

	/**
	 * @return the intervalOfCheckTimeoutEvent
	 */
	public long getIntervalOfCheckTimeoutEvent() {
		return intervalOfCheckTimeoutEvent;
	}

	/**
	 * @param intervalOfCheckTimeoutEvent the intervalOfCheckTimeoutEvent to set
	 */
	public synchronized void setIntervalOfCheckTimeoutEvent(
			long intervalOfCheckTimeoutEvent) {
		this.intervalOfCheckTimeoutEvent = intervalOfCheckTimeoutEvent;
		this.rescheduleCheckTimeoutEvent();
	}

	/**
	 * @return the intervalOfAdjustThreadPoolCoreSize
	 */
	public long getIntervalOfAdjustExecutorCorePoolSize() {
		return intervalOfAdjustExecutorCorePoolSize;
	}

	/**
	 * @param intervalOfAdjustThreadPoolCoreSize the intervalOfAdjustThreadPoolCoreSize to set
	 */
	public synchronized void setIntervalOfAdjustExecutorCorePoolSize(
			long intervalOfAdjustThreadPoolCoreSize) {
		this.intervalOfAdjustExecutorCorePoolSize = intervalOfAdjustThreadPoolCoreSize;
		this.rescheduleAdjustThreadPoolCoreSize();
	}
	
    private static void updateLargestCount(final AtomicInteger largestCount, 
    		final int currentCount) {
    	int cnt = largestCount.get();
		boolean modified = false;
    	while ( !modified && cnt < currentCount ) {
    		modified = largestCount.compareAndSet(cnt, currentCount);
    		cnt = largestCount.get();
    	}
    }
    
    private void enlargePoolSize(int currentPoolSize) {
		if ( null != workService ) {
			if ( currentPoolSize > workService.getCorePoolSize() ) {
				synchronized (workService) {
					//	double check for more effective
					if ( currentPoolSize > workService.getCorePoolSize() ) {
						workService.setCorePoolSize(currentPoolSize);
					}
				}
			}
		}
    }
    
    private void reducePoolSize(int currentPoolSize) {
		if ( null != workService ) {
			if ( currentPoolSize < workService.getCorePoolSize() ) {
				synchronized (workService) {
					if ( currentPoolSize < workService.getCorePoolSize() ) {
						workService.setCorePoolSize(currentPoolSize);
					}
				}
			}
		}
    }
    
    private	void incActiveJobCount() {
    	int nowCount = activeJobCount.incrementAndGet();
    	updateLargestCount(largestActiveJobCount, nowCount);
    	updateLargestCount(periodLargestActiveJobCount, nowCount);
    	enlargePoolSize(Math.min( nowCount, executorPoolSizeLimit ));
    }
    
    private	void decActiveJobCount() {
		activeJobCount.decrementAndGet();			
    }
    
    private void resetLargestActiveJobCount() {
    	try {
	    	this.periodLargestActiveJobCount.set(this.activeJobCount.get());
	    	reducePoolSize(periodLargestActiveJobCount.get());
    	}
    	catch (Exception e) {
    		logger.error("resetLargestActiveJobCount:", e);
    	}
	}

	private	void rescheduleCheckTimeoutEvent() {
		try {
			if ( null != futureOfCheckTimeoutEvent ) {
				futureOfCheckTimeoutEvent.cancel(true);
			}
			futureOfCheckTimeoutEvent = this.innerService.scheduleWithFixedDelay(
					new Runnable(){
						public void run() {
							doCheckTimeoutEvents();
						}}, 
				this.intervalOfCheckTimeoutEvent, 
				this.intervalOfCheckTimeoutEvent, 
				TimeUnit.MILLISECONDS);
		}
		catch (Exception e) {
			logger.error("rescheduleCheckTimeoutEvent:", e);
		}
	}

	private	void rescheduleAdjustThreadPoolCoreSize() {
		try {
			if ( null != futureOfAdjustExecutorCorePoolSize ) {
				futureOfAdjustExecutorCorePoolSize.cancel(true);
			}
			futureOfAdjustExecutorCorePoolSize = this.innerService.scheduleWithFixedDelay(
					new Runnable(){
						public void run() {
							resetLargestActiveJobCount();
						}}, 
					this.intervalOfAdjustExecutorCorePoolSize, 
					this.intervalOfAdjustExecutorCorePoolSize, 
					TimeUnit.MILLISECONDS);
		}
		catch (Exception e) {
			logger.error("rescheduleAdjustThreadPoolCoreSize:", e);
		}
	}
	
    public	static class LongCounter {
    	final AtomicLong	atomicCounter;
    	
    	public LongCounter(AtomicLong al) {
    		atomicCounter = al;
    	}
    	
    	public long getDealCount() {
    		return	atomicCounter.get();
    	}
    }
    
	private AtomicLong createAndRegisterEndReasonCounter(final Object reason) {
		AtomicLong al = new AtomicLong(0);
		registerMBean(
				createObjectNameForDeal(reason, "category=all"), 
				new LongCounter(al) );
		
		return	al;
	}
	
	private void countDealForEndReason(final Object reason) {
		String key = reason.toString();
		AtomicLong counter = erCounters.get(key);
		if ( null == counter ) {
			counter = createAndRegisterEndReasonCounter(reason);
			erCounters.put(key, counter);
		}
		counter.incrementAndGet();
	}
	
	private static Class<?> getIndicateType(final Class<?> cls) {
		final IndicateInterface anno = cls.getAnnotation(IndicateInterface.class);
		if ( null != anno ) {
			return	anno.type();
		}
		
		return	null;
	}
	
	private class CountersGetterImpl implements InvocationHandler {
		
		private	final Map<Object, AtomicLong> counters;
		
		public CountersGetterImpl(final Map<Object, AtomicLong> counters) {
			this.counters = counters;
		}
		
		public Object invoke(final Object proxy, final Method method, final Object[] args)
				throws Throwable {
	        final String methodName = method.getName();
	        final Class<?> returnType = method.getReturnType();
	
	        /* Inexplicably, InvocationHandler specifies that args is null
	           when the method takes no arguments rather than a
	           zero-length array.  */
	        final int nargs = (args == null) ? 0 : args.length;
	
	        if (methodName.startsWith("get")
	            && methodName.length() > 3
	            && nargs == 0
	            && (returnType.equals(long.class)
	            	|| returnType.equals(Long.class) ) ) {
	        	
	        	final String key;
	        	
	        	Indicator indicator = method.getAnnotation(Indicator.class);
	        	if ( null != indicator ) {
	        		key = indicator.key();
	        	}
	        	else {
	        		key = methodName.substring(3);
	        	}
	        	
				try {
					return	innerService.submit(new Callable<Long>() {
						public Long call() throws Exception {
							AtomicLong ret = counters.get(key);
							return	null != ret ? ret.get() : 0;
						}}).get();
				} catch (Exception e) {
					logger.error("DealTimeInvocationHandler.invoke:", e);
					return	0;
				}
	        }
	
	        logger.warn("Invalid method {} while valid is : getXXX"
	        		+"\r\n Or invalid parameters count {}", 
	        		methodName, nargs );
			return 0;
		}
		
	}
	
	private ERTuple createAndRegisterEndReasonTuple(
			final Transformer	transformer, final Object reason, final String category) {
		final ERTuple tuple = new ERTuple();
		tuple.trans = transformer;
		Class<?> cls = getIndicateType(tuple.trans.getClass());
		if ( null != cls ) {
			registerMBean(
					createObjectNameForDeal(reason, "category=" + category), 
					Proxy.newProxyInstance(
							cls.getClassLoader(), 
							new Class<?>[]{cls}, 
							new CountersGetterImpl( tuple.counter ) ) );
		}
		else {
			logger.error("can't get IndicateType for [" + tuple.trans.getClass() + "]");
		}
		
		return	tuple;
	}
	
    //TODO
    private void countTimeForEndReason(final Transformer defaultTransformer, 
    		final Map<String, Long> stateLiveTime) {
        for (String stateName : stateLiveTime.keySet()) {
            ERTuple tuple = erTuplesOfStateAwareTTL.get(stateName);
            if (tuple == null) {
                tuple = createAndRegisterEndReasonTuple(defaultTransformer, "stateAwareTTL,state="+stateName, "ttl");
                erTuplesOfStateAwareTTL.put(stateName, tuple);
            }

            final Transformer trans = tuple.trans;
            if (null != trans) {
                final Object key = trans.transform(stateLiveTime.get(stateName));
                if (null != key) {
                    AtomicLong count = tuple.counter.get(key);
                    if (null == count) {
                        count = new AtomicLong(0);
                        tuple.counter.put(key, count);
                    }
                    count.incrementAndGet();
                }
            }
        }
        stateLiveTime.clear();
    }

	private void countTimeForEndReason(
			long 		value,
			Object 		reason, 
			Transformer	defaultTransformer,				
			Map<String, ERTuple> tuples, 
			String		category) {
		
		String	keyOfTuple = reason.toString();
		ERTuple tuple = tuples.get(keyOfTuple);
		
		if ( null == tuple ) {
			tuple = createAndRegisterEndReasonTuple(defaultTransformer, reason, category);
			tuples.put(keyOfTuple, tuple);
		}
		
		Transformer	trans = tuple.trans;
		if ( null != trans ) {
			final Object key = trans.transform(value);
			if ( null != key ) {
				AtomicLong count = tuple.counter.get(key);
				if ( null == count ) {
					count = new AtomicLong(0);
					tuple.counter.put(key, count);
				}
				count.incrementAndGet();
			}
		}
	}

	private void doCountDeal(Object reason, long timeToLive, long timeToActive,Map<String, Long> stateLiveTime) {
		
		if ( null == reason ) {
			reason = "";
		}
		
		countDealForEndReason(reason);
		
		//	ttl 
		countTimeForEndReason(
				timeToLive,
				reason, 
				transformerOfTTL, 
				erTuplesOfTTL, 
				"ttl");

		// 	tta
		countTimeForEndReason(
				timeToActive,
				reason, 
				transformerOfTTA, 
				erTuplesOfTTA, 
				"tta");

        //stateAware ttl
        countTimeForEndReason(transformerOfTTA,stateLiveTime);
	}
	
	private void doResetDealTimeCounter() {
//		dealTimeToLiveCounter.clear();
//		dealTimeToActiveCounter.clear();
		//	TODO
		//	clear endreason counter
	}
	
	private String doGetDealTimeCountAsString() {
		StringBuilder sb = new StringBuilder();
		
		sb.append("timeToLive:\r\n");
//		for (Map.Entry<Object, AtomicLong> entry : dealTimeToLiveCounter.entrySet()) {
//			sb.append(entry.getKey());
//			sb.append(":\t");
//			sb.append(entry.getValue().get());
//			sb.append(" \r\n");
//		}
//		
//		sb.append("timeToActive:\r\n");
//		for (Map.Entry<Object, AtomicLong> entry : dealTimeToActiveCounter.entrySet()) {
//			sb.append(entry.getKey());
//			sb.append(":\t");
//			sb.append(entry.getValue().get());
//			sb.append(" \r\n");
//		}
		
		return	sb.toString();
	}
	
	private void doResetDealEndReasonCounter() {
//		erCounters.clear();
	}
	
	private String doGetDealEndReasonCountAsString() {
		final StringBuilder sb = new StringBuilder();
		
		for (Map.Entry<Object, AtomicLong> entry : erCounters.entrySet()) {
			sb.append(entry.getKey());
			sb.append(":\t");
			sb.append(entry.getValue().get());
			sb.append(" \r\n");
		}
		
		return	sb.toString();
	}
	
	private void updateAllDealStatistics(final SmartContext ctx) {
		final long timeToLive = ctx.getTimeToLive();
		final long timeToActive = ctx.getTimeToActive();
		
		if ( timeToLive > this.dealLargestTimeToLive ) {
			this.dealLargestTimeToLive = timeToLive;
		}
		if ( timeToLive < this.dealSmallestTimeToLive ) {
			this.dealSmallestTimeToLive = timeToLive;
		}
		
		if ( timeToActive > this.dealLargestTimeToActive ) {
			this.dealLargestTimeToActive = timeToActive;
		}
		if ( timeToActive < this.dealSmallestTimeToActive ) {
			this.dealSmallestTimeToActive = timeToActive;
		}
		
		dealTotalTimeToLive.addAndGet(timeToLive);
		dealTotalTimeToActive.addAndGet(timeToActive);
		final Object reason = ctx.getEndReason();

        final Map<String, Long> stateLiveTime = ctx.getStateLiveTime();

		innerService.submit(new Runnable() {
			public void run() {
				doCountDeal(reason, timeToLive, timeToActive,stateLiveTime);
			}});
	}

	private void removeContext(final SmartContext ctx) {
		final Object key = ctx.getKey();
		
		this.keyFactory.removeKey(key);
		if ( null != this.contextHolder.remove(key) ) {
			//	移除操作有效
			this.totalJobCount.decrementAndGet();
		}
		
		try {
			final List<Object> pendings = ctx.destroy();
			if ( null != pendings ) {
				//	ctx 已经结束，但还存在未处理的event
				//	将其重新放入FSM中，触发相关处理流程
				if ( logger.isInfoEnabled() ) {
					logger.info("ctx {} destroy and {} events left, try re-accept by fsm {}",
							new Object[]{ctx, pendings.size(), this});
				}
				for ( Object event : pendings) {
					if ( logger.isDebugEnabled() ) {
						logger.debug("fsm {} re-accept event {}", this, event);
					}
					this.acceptEvent(event);
				}
			}
		}
		catch (Exception e) {
			logger.error("[" + name + "]removeContext:", e);
		}
		
		dealCompletedCount.incrementAndGet();
		updateAllDealStatistics(ctx);
	}

	private State getStateOf(String stateName) {
		return	this.stateRepository.find(stateName);
	}
	
	private void fireStateChanged(final FSMContext ctx, final String fromState, final String toState, final Object event) {
		if ( logStateChanged && null != eventBus && null != eventStateChanged ) {
			final Object snapshot = ctx.getSnapshot();
			
			if ( null != snapshot ) {
				try {
					PropertyUtils.setProperty(snapshot, "fsm", name);
					PropertyUtils.setProperty(snapshot, "fromState", fromState);
					PropertyUtils.setProperty(snapshot, "toState", toState);
					PropertyUtils.setProperty(snapshot, "eventRecved", 
							null != event ? event.toString() : "");
					PropertyUtils.setProperty(snapshot, "time", new Date());
				} catch (IllegalAccessException e) {
					logger.error("fireStateChanged:", e);
				} catch (InvocationTargetException e) {
					logger.error("fireStateChanged:", e);
				} catch (NoSuchMethodException e) {
					logger.error("fireStateChanged:", e);
				}
				
				eventBus.fireEvent(this.eventStateChanged, snapshot);
			}
		}
	}
	
	private	void initSnapshot(final MutableFSMContext ctx) {
		if ( this.logStateChanged && (null != this.snapshotFactory) ) {
			try {
				ctx.setSnapshot(snapshotFactory.transform(ctx));
			}
			catch (Exception e) {
				logger.error("[" + name + "]initSnapshot:", e);
			}
		}
	}
	
	private void initContext(final SmartContext ctx, final Object event) {
		final State state = getStateOf(ctx.getCurrentState());
		if ( null == state ) {
			logger.warn("Can not found valid state for [" + ctx.getCurrentState() 
					+ "] , remove context [" + ctx + "]");
			removeContext(ctx);
		}
		else {
			
			initSnapshot(ctx);
			
			boolean ret = false;
			
			try {
				ret = state.enter(this, ctx);
			}
			catch (Exception e) {
				logger.error( ctx.getCurrentState() + ".enter:", e);
			}
			
			if ( !ret ) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("enter return false, ctx [" + ctx + "] will end normally.");
				}
				removeContext(ctx);
				return;
			}
			fireStateChanged(ctx, "", ctx.getCurrentState(), event);
			
			initSnapshot(ctx);
			
			dispatchEvent(ctx, event);
		}
	}
	
	private void dispatchEvent(final SmartContext ctx, final Object event) {
        final long stateBegin = System.currentTimeMillis();
		final State currentState = getStateOf(ctx.getCurrentState());
		if ( null == currentState ) {
			logger.error("Internal Error: Can not found valid state for [" + ctx.getCurrentState() 
					+ "] , remove context [" + ctx + "]");
			removeContext(ctx);
		}
		
		String nextStateName = null;

		try {
			nextStateName = currentState.acceptEvent(this, ctx, event);
		}
		catch (Exception e) {
			logger.error(ctx.getCurrentState() + ".acceptEvent:", e);
		}
		
		if ( null == nextStateName ) {
			// this context's FSM state reach end
			try {
				currentState.leave(this, ctx);
			}
			catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".leave:", e);
			}
			
			if ( logger.isDebugEnabled() ) {
				logger.debug("ctx [" + ctx + "] will end normally.");
			}
			
			fireStateChanged(ctx, ctx.getCurrentState(), "", event);
			removeContext(ctx);
		}
		else if ( nextStateName.equals(ctx.getCurrentState()) ) {
			// no change
		}
		else {
			final State nextState = getStateOf(nextStateName);
			if ( null == nextState ) {
				logger.error("Internal Error: Can not found valid state for [" + nextStateName
						+ "] , just ignore");
			}
			else {
				// next state is valid
				try {
					currentState.leave(this, ctx);
				}
				catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".leave:", e);
				}
                //TODO 上个状态完毕,进入下个状态
                if (this.getInitStateName().equalsIgnoreCase(ctx.getCurrentState())) {
                    ctx.setStateLiveTime(ctx.getCurrentState(), System.currentTimeMillis() - stateBegin);
                } else {
                    ctx.setStateLiveTime(ctx.getCurrentState(), System.currentTimeMillis() - ctx.getLastModify());
                }

				final String prevStateName = ctx.getCurrentState();
				ctx.setCurrentState(nextStateName);
				
				boolean ret = false;
				try {
					ret = nextState.enter(this, ctx);
				}
				catch (Exception e) {
					logger.error( ctx.getCurrentState() + ".enter:", e);
				}
				
				fireStateChanged(ctx, prevStateName, ctx.getCurrentState(), event);
				initSnapshot(ctx);
				
				if ( !ret ) {
					if ( logger.isDebugEnabled() ) {
						logger.debug("enter return false, ctx [" + ctx + "] will end normally.");
					}
					
					removeContext(ctx);
				}
			}
		}
		
		try {
			ctx.endOfDispatchEvent(new Closure(){
	
				public void execute(final Object input) {
					dispatchEvent(ctx, input);
				}});
		}
		catch (Exception e) {
			logger.error("ctx.endOfDispatchEvent{}, try end ctx", e);
			removeContext(ctx);
		}
	}
	
	private void incDealBypass() {
		dealBypassCount.incrementAndGet();
	}
	
	private void incDealHandledStart() {
		dealHandledCount.incrementAndGet();
	}
	
	private boolean checkIfExceedLimit(Object event) {
		if ( this.checkBusy 
				&& this.activeJobCount.get() >= this.busyThreshold ) {
			
			incDealBypass();
			
			//	超过CTX限制值
			if ( null != busyHandler ) {
				try {
					busyHandler.execute(event);
				}
				catch (Exception e) {
					logger.error(busyHandler.toString() + ".execute:", e);
				}
			}
			else {
				logger.error("reach active ctx limit, and missing BusyHandler.");
			}
			return	true;
		}
		return	false;
	}

	public void acceptEvent(final Object event) {
		final Object key = keyFactory.transformToKey(event);
		
		if ( null == key ) {
			logger.warn("event " + event + " can not transform key, just ignore");
			return;
		}
		//	lookup key in contextHolder
		final SmartContext ctx = contextHolder.get(key);
		if ( null == ctx ) {
			if ( !this.initeventAcceptor.evaluate(event)) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("event [" + event + "] not acceptable, just ignore");
				}
				return;
			}
			
			if ( checkIfExceedLimit(event) ) {
				if ( logger.isDebugEnabled() ) {
					logger.debug("FSM[" + name + "] busy, event[" + event + "] has been dropped.");
				}
				return;
			}
			
			final SmartContext initCtx = createSmartContext();
			if ( null == initCtx ) {
				logger.warn("Can not create context for event " + event + " , just ignore");
				return;
			}
			
			incDealHandledStart();
			
			//	init ctx, and continue
			initCtx.setService(workService);
			initCtx.setKey(key);
			initCtx.setActived();
			
			if ( null == contextHolder.put(key, initCtx) ) {
				// add new context
				totalJobCount.incrementAndGet();
			}
			
			workService.submit(new Runnable(){

				public void run() {
					initContext(initCtx, event);
				}});
		}
		else {
			try {
				ctx.processEvent(event, new Closure(){
	
					public void execute(final Object input) {
						dispatchEvent(ctx, input);
					}} );
			}
			catch (Exception e) {
				logger.error("ctx.processEvent error, try end ctx", e);
				removeContext(ctx);
			}
		}
	}

	//	in timerExecService thread
	private void doScheduleTimeout(final TimeoutEvent event, final long delay) {
		timeoutHolder.put(event, System.currentTimeMillis() + delay);
	}
	
	//	in timerExecService thread
	private void doCancelTimeout(TimeoutEvent event) {
		if ( null == timeoutHolder.remove(event) ) {
			if ( logger.isWarnEnabled() ) {
				logger.warn("Can not found matched timeout event[" + event + "], maybe has been emited.");
			}
		}
		else {
			if ( logger.isDebugEnabled() ) {
				logger.debug("cancel timeout event[" + event + "] succeed.");
			}
		}
	}
	
	//	in timerExecService thread
	private void doCheckTimeoutEvents() {
		final long now = System.currentTimeMillis();
		
		try {
			final Set<Map.Entry<TimeoutEvent, Long>> entrySet = timeoutHolder.entrySet();
			
			final Iterator<Map.Entry<TimeoutEvent, Long>> iter = entrySet.iterator();
			while ( iter.hasNext() ) {
				final Map.Entry<TimeoutEvent, Long> entry = iter.next();
				if ( entry.getValue() <= now ) {
					//	condition matched, emit timeout event
					iter.remove();
					this.acceptEvent(entry.getKey());
				}
			}
		}
		catch (Exception e) {
			logger.error("doCheckTimeoutEvents:", e);
		}
	}
	
	public Runnable scheduleTimeout(final TimeoutEvent event, final long delay) {
		this.innerService.submit(new Runnable(){

			public void run() {
				doScheduleTimeout(event, delay);
			}});
		
		return new Runnable(){

			public void run() {
				innerService.submit(new Runnable(){

					public void run() {
						doCancelTimeout(event);
					}});
			}};
	}

	public class DashboardImpl implements FSMDashboard {
		public long getDealAverageTimeToActive() {
			int completed = dealCompletedCount.get();
			return completed > 0 ? dealTotalTimeToActive.get() / completed : -1;
		}

		public long getDealAverageTimeToLive() {
			int completed = dealCompletedCount.get();
			return completed > 0 ? dealTotalTimeToLive.get() / completed : -1;
		}

		public long getDealBypassCount() {
			return dealBypassCount.get();
		}

		public long getDealCompletedCount() {
			return dealCompletedCount.get();
		}

		public long getDealHandledCount() {
			return dealHandledCount.get();
		}

		public long getDealLargestTimeToActive() {
			return dealLargestTimeToActive;
		}

		public long getDealLargestTimeToLive() {
			return dealLargestTimeToLive;
		}

		public long getDealSmallestTimeToActive() {
			return dealSmallestTimeToActive;
		}

		public long getDealSmallestTimeToLive() {
			return dealSmallestTimeToLive;
		}

		public int getExecutorActiveThreadCount() {
			return workService.getActiveCount();
		}

		public long getExecutorCompletedTaskCount() {
			return workService.getCompletedTaskCount();
		}

		public int getExecutorCorePoolSize() {
			return workService.getCorePoolSize();
		}

		public int getExecutorCurrentPoolSize() {
			return workService.getPoolSize();
		}

		public long getExecutorHandledTaskCount() {
			return workService.getTaskCount();
		}

		public int getExecutorLargestPoolSize() {
			return workService.getLargestPoolSize();
		}

		public int getExecutorMaximumPoolSize() {
			return workService.getMaximumPoolSize();
		}

		public int getExecutorPendingTaskCount() {
	    	if ( null != workService ) {
	    		return	workService.getQueue().size();
	    	}
	    	return	-1;
		}

		public int getJobActiveCount() {
			return	activeJobCount.get();
		}

		public int getJobInactiveCount() {
			return getJobTotalCount() - getJobActiveCount();
		}
		
		public int getJobTotalCount() {
			return	totalJobCount.get();
		}

		public int getJobLargestActiveCount() {
	    	return	largestActiveJobCount.get();
		}

		public int getJobPeriodLargestActiveCount() {
	    	return	periodLargestActiveJobCount.get();
		}

		public int getJobTimeoutEventCount() {
			return	timeoutHolder.size();
		}
		
		public String[] getJobsDetail() {
			final Set<Map.Entry<Object, SmartContext>> ctxs = contextHolder.entrySet();
			
			final List<String> ret = new ArrayList<String>();
			final SimpleDateFormat formater = new SimpleDateFormat("yyMMdd-HH:mm:ss.SSS");
			
			for ( Map.Entry<Object, SmartContext> entry : ctxs ) {
				final StringBuilder sb = new StringBuilder();
				
				sb.append(entry.getKey().toString());
				sb.append(":[");
				
				FSMContext ctx = entry.getValue();
				sb.append(ctx.getCurrentState());
				sb.append("]create:");
				sb.append(formater.format( new Date(ctx.getCreateTime())) );
				sb.append("/last:");
				sb.append(formater.format( new Date(ctx.getLastModify())) );
				ret.add(sb.toString());
			}
			
			return	ret.toArray(new String[0]);
		}

		public int getInnerServiceTaskCount() {
			if ( null != innerService ) {
				return	innerService.getQueue().size();
			}
			return	-1;
		}

		public void resetDealTimeExtremum() {
		    dealLargestTimeToLive = -1;
		    dealLargestTimeToActive = -1;
		    dealSmallestTimeToLive = Integer.MAX_VALUE;
		    dealSmallestTimeToActive = Integer.MAX_VALUE;
		}

		public String getDealTimeCountAsString() {
			try {
				return	innerService.submit(new Callable<String>() {
					public String call() throws Exception {
						return doGetDealTimeCountAsString();
					}}).get();
			} catch (Exception e) {
				logger.error("Dashboard.getDealTimeCountAsString:", e);
				return	null;
			}
		}

		public void resetDealTimeCounter() {
			try {
				innerService.submit(new Runnable() {
					public void run() {
						doResetDealTimeCounter();
					}}).get();
			} catch (Exception e) {
				logger.error("Dashboard.resetDealTimeCounter:", e);
			}
		}

		public String getDealEndReasonCountAsString() {
			try {
				return	innerService.submit(new Callable<String>() {
					public String call() throws Exception {
						return doGetDealEndReasonCountAsString();
					}}).get();
			} catch (Exception e) {
				logger.error("Dashboard.getDealEndReasonCountAsString:", e);
				return	null;
			}
		}

		public void resetDealEndReasonCounter() {
			try {
				innerService.submit(new Runnable() {
					public void run() {
						doResetDealEndReasonCounter();
					}}).get();
			} catch (Exception e) {
				logger.error("Dashboard.resetDealEndReasonCounter:", e);
			}
		}
	}

	public FSMDashboard createDashboard() {
		return	new DashboardImpl();
	}

	/**
	 * @return the mbeanExport
	 */
	public MBeanExporter getMbeanExporter() {
		return mbeanExporter;
	}

	/**
	 * @param mbeanExport the mbeanExport to set
	 */
	public void setMbeanExporter(MBeanExporter mbeanExport) {
		this.mbeanExporter = mbeanExport;
	}

	/**
	 * @return the objectNamePrefix
	 */
	public String getObjectNamePrefix() {
		return objectNamePrefix;
	}

	/**
	 * @param objectNamePrefix the objectNamePrefix to set
	 */
	public void setObjectNamePrefix(final String objectNamePrefix) {
		this.objectNamePrefix = objectNamePrefix;
	}

	/* (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return this.name;
	}

    protected abstract SmartContext createFSMContext();
	
	private SmartContext createSmartContext() {
		final SmartContext	ctx = createFSMContext();
		
		ctx.setCurrentState(this.initStateName);
		if ( null != eventOfDestroyContext ) {
			ctx.setEventOfDestroy(eventOfDestroyContext);
		}
		ctx.setStatusListener(ctxStatusListener);
		
		return ctx;
	}

}
